package plugin.voldemort.lookup;

import org.pentaho.di.core.Const;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.exception.KettleStepException;
import org.pentaho.di.core.row.RowDataUtil;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.core.row.ValueMetaInterface;
import org.pentaho.di.i18n.BaseMessages;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.*;

import voldemort.client.ClientConfig;
import voldemort.client.SocketStoreClientFactory;
import voldemort.client.StoreClientFactory;


public class LookupStep extends BaseStep implements StepInterface {

	private LookupStepData data;
	private LookupStepMeta meta;
	
	private static Class<?> PKG = LookupStep.class; // for i18n purposes
	
	public LookupStep(StepMeta s, StepDataInterface stepDataInterface, int c, TransMeta t, Trans dis) {
		super(s, stepDataInterface, c, t, dis);
	}

	public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException {

		meta = (LookupStepMeta) smi;
		data = (LookupStepData) sdi;

		Object[] r = getRow(); // get row, blocks when needed!
		if (r == null) // no more input to be expected...
		{
			setOutputDone();
			return false;
		}

		if (first) {
			
			first = false;
			
			// the size of the incoming rows 
			data.inputSize = getInputRowMeta().size();
			
			// determine output field structure
			data.outputRowMeta = (RowMetaInterface) getInputRowMeta().clone();
			meta.getFields(data.outputRowMeta, getStepname(), null, null, this);
		}
		
		// generate output row, make it correct size
		Object[] outputRow = RowDataUtil.resizeArray(r, data.outputRowMeta.size());
		
		return true;
	}

	public boolean init(StepMetaInterface smi, StepDataInterface sdi) {
		meta = (LookupStepMeta) smi;
		data = (LookupStepData) sdi;
					
	    // get a voldemort client
	    String bootstrapUrl = "tcp://"+environmentSubstitute(meta.getVoldemortHost())+":"+environmentSubstitute(meta.getVoldemortPort());
	    StoreClientFactory factory = new SocketStoreClientFactory(new ClientConfig().setBootstrapUrls(bootstrapUrl));
	    
	    if (data.voldemortClient == null){
	    	return false;
	    }
	       
		return super.init(smi, sdi);
	}

	public void dispose(StepMetaInterface smi, StepDataInterface sdi) {
		meta = (LookupStepMeta) smi;
		data = (LookupStepData) sdi;
		
		data.voldemortClient = null;

		super.dispose(smi, sdi);
	}

}
